package com.citycloud.ccuap.tc.sync.handle;

import com.alibaba.fastjson.JSON;
import com.citycloud.ccuap.commons.application.ApplicationContextProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

public class KafkaHandle implements MessageHandleFactory {

    private static Logger logger = LoggerFactory.getLogger(KafkaHandle.class);
    private final KafkaTemplate<String, Object> kafkaTemplate = (KafkaTemplate<String, Object>) ApplicationContextProvider.getBean("kafkaTemplate");

    @Override
    public void sendMessage(String topic, Object message) {
        String messageJson = JSON.toJSONString(message);
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, messageJson);

        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {

            @Override
            public void onFailure(Throwable throwable) {
                logger.info("Produce: The message failed to be sent:{}", throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                logger.info("Produce: _+_+_+_+_+_+_+ result: {}", stringObjectSendResult.toString());
            }
        });
    }
}
